package org.example.sql;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.example.data.WaterSensor;

import static org.apache.flink.table.api.Expressions.$;

public class TableStreamDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<WaterSensor> sensorDS = env.fromElements(new WaterSensor("s1", 1L, 1),
                new WaterSensor("s2", 2L, 2),
                new WaterSensor("s3", 3L, 3),
                new WaterSensor("s4", 4L, 4),
                new WaterSensor("s4", 5L, 5));

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //指定字段
        //Table table = tableEnv.fromDataStream(sensorDS,$("id"));
        Table sensorTable = tableEnv.fromDataStream(sensorDS);
        tableEnv.createTemporaryView("sensor", sensorTable);

        Table resultTable = tableEnv.sqlQuery("select id,sum(vc) as sumVC from sensor group by id;");
        Table filterTable = tableEnv.sqlQuery("select * from sensor where id='s1'");

        //追加流
        tableEnv.toDataStream(filterTable, WaterSensor.class).print("filter");

        //更新流
        tableEnv.toChangelogStream(resultTable).print("result");
        //只要代码中调用了DataStreamAPI就要执行env.execute，否则不需要。
        env.execute("TableStream");
    }
}
